(ns data2.core
  (:gen-class)
  (:use clojure.data.csv
        clojure.java.io)
  (:require [clj-time.core :as t]
            [clj-time.format :as tf]
            [clj-time.coerce :as tc])
  (:import [java.util Random]))


(def time-formatter (tf/formatter "HH:mm:ss"))

(def date-formatter (tf/formatter "yyyyMMdd"))
(def range-formatter (tf/formatter "yyyy-MM-dd"))


(defn trans-range-date [d]
  (tf/unparse date-formatter (tf/parse range-formatter d)))


(defonce working-agents (atom nil))


(defn parse-time [date-str]
  (tf/parse time-formatter date-str))


(defn unparse-time [date]
  (tf/unparse time-formatter date))


(def day-start-time (parse-time "09:30:00"))
(def day-end-time (parse-time "16:00:01"))


(defn process-daily-data
  "处理天数据"
  [data minutes-per-sample]
  (let [total (count data)
        first-minute (parse-time (nth (first data) 2))
        last-minute (parse-time (nth (last data) 2))
        minutes (/ (- (tc/to-long last-minute) (tc/to-long first-minute)) (* 1000 60))
        number-of-samples (+ (int (/ minutes minutes-per-sample)) 1)
        sample-skip (int (/ total number-of-samples))
        number-of-extra (- total (* number-of-samples sample-skip))
        random-generater (Random. 1)]
    (->> (for [n (range 10)]
           (let [number-to-drop (int (* (. random-generater nextDouble) number-of-extra))
                 data-seq (drop number-to-drop data)]
             (->> data-seq
                  (partition sample-skip)
                  (map first)
                  (map (fn [[_ d t p _ _]]
                         [d t p])))))
         (apply (partial map (fn [& groups]
                               (apply concat groups))))
         (map-indexed (fn [index row]
                        (concat row (list (unparse-time (tc/from-long (+ (tc/to-long day-start-time) (* index minutes-per-sample (* 1000 60))))))))))))


(defn process-data
  "处理数据data，时间间隔为minutes-per-sample"
  [data minutes-per-sample range-map]
  (let [valid-data (->> data
                        (filter (fn [data]
                                  (let [t (parse-time(nth data 2))]
                                    (t/within?
                                     (t/interval
                                      day-start-time day-end-time)
                                     t))))
                        (filter (fn [[_ d _ ps _ _]]
                                  (let [p (Double/parseDouble ps)
                                        r (range-map d)
                                        ]
                                    (if (nil? r) false
                                        (let [t (Double/parseDouble (:top r))
                                              b (Double/parseDouble (:bottom r))]
                                          (and (<= p t)
                                               (>= p b)))))))
                        )
        ;; 将数据按天分组
        daily-data (partition-by #(nth % 1) valid-data)]
    (map (fn [data]
           (process-daily-data data minutes-per-sample)) daily-data)))


(defn process
  "处理y年份的数据，时间间隔为skip分钟"
  [y skip]
  (let [in-file (str y ".csv")
        out-file (str "data/" y "-" skip ".csv")]
    (with-open [in (reader in-file)
                out (writer out-file)
                rin (reader "range.csv")]
      (write-csv out [["date1" "time1" "price1"
                       "date2" "time2" "price2"
                       "date3" "time3" "price3"
                       "date4" "time4" "price4"
                       "date5" "time5" "price5"
                       "date6" "time6" "price6"
                       "date7" "time7" "price7"
                       "date8" "time8" "price8"
                       "date9" "time9" "price9"
                       "date10" "time10" "price10"
                       "time"]])
      (let [data (drop 1 (read-csv in))
            range-map (reduce (fn [result [d _ h l _ _ _]]
                                (assoc result (trans-range-date d) {:top h :bottom l})) {} (drop 1 (read-csv rin)))
            result (process-data data skip range-map)]
        (reduce (fn [_ v]
                  (write-csv out v)) nil result)))))


(defn show-status
  [agents]
  (doseq [r @(delay (partition-all 5 agents))]
    (doseq [a r]
      (let [[y skip state] @a] (print y skip state "|\t")))
    (println)))


(defn show-status-repeatly
  [agents]
  (doseq [i (repeat 0)]
    (Thread/sleep 1000)
    (#'show-status agents)
    (println "-----------------------------------------------------------")))


(defn status-report
  []
  (show-status-repeatly @working-agents))

(defn start-process
  []
  (set-agent-send-executor!
   ;; set the agent send executor, set the number of threads we shall use to process the data.
   (java.util.concurrent.ThreadPoolExecutor.
    2 2 0 java.util.concurrent.TimeUnit/MINUTES (java.util.concurrent.LinkedBlockingQueue.)))

  (swap! working-agents (fn [_]
                          (for [skip [1 3 5 10 30]
                                y (range 2013 2014)]
                            (let [process-agent (agent [y skip "r"])]
                              (send process-agent (fn [[y skip _]]
                                                    (try
                                                      (process y skip)
                                                      [y skip "d"]
                                                      (catch Throwable e
                                                        [y skip "e" e]))))

                              process-agent)))))

(defn -main [& args]
  (start-process)
  (status-report)
  ;;(process 2013 180)
  )
